package cn.ipanel.bigdata.job.ods

import cn.ipanel.bigdata.boot.date.Day
import cn.ipanel.bigdata.boot.period.Period
import cn.ipanel.bigdata.dw.dim.phoenix.{T_DEVICE => D}
import cn.ipanel.bigdata.dw.ods
import cn.ipanel.bigdata.dw.ods.{T_REPORT_BY_HOUR, T_REPORT => T}
import cn.ipanel.bigdata.utils.Dictionary
import cn.ipanel.bigdata.utils.Dictionary.BusState
import cn.ipanel.bigdata.utils.Util.transNullAndEmpryAsString
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.IntegerType

/**
 * @author lzz
 * @projectName iBigdata
 * @date 2023/03/21 15:33
 * @description: 工具，用于将一个用户的上报，放在一个executor处理
 */
class Report extends AbsOdsJob {

  override def optTargetTable: ods.AbsOdsHive = T

  override def source(): DataFrame = {
    ods.T_HDFS_LOG.readToTable(_DAY)
      // 当天的日志，第二天可能会出现，这里先这样做，看看后面是否要调整
      .union(ods.T_HDFS_LOG.readToTable(_DAY.next()))
  }

  override def intent(data: DataFrame, day: Day): DataFrame = {
    data.filter(col(T.F_DATE) === lit(_DAY.toDate))
  }

//  override def join(data: DataFrame, day: Day): DataFrame = {
//    val df = data.join(
//                      // 这种写法，在local模式下，能补充f_region字段，但是服务器上（yarn模式）下不能，暂时不知道原因
//                      Area.IdSupplier
//                        .supplyFRegion(spark,
//                          D.load
//                            .selectExpr(D.SELECT_FIELDS: _*)
//                            .withColumnRenamed(D.F_MAC, T.F_DEVICE_ID)
//                            .withColumnRenamed(D.F_MANUFACTURER_OUI, T.F_MANUFACTURE)
//                            .withColumnRenamed(D.F_SUPPORT_LICENSOR, T.F_CP)
//                        )
//      , Seq(T.F_DEVICE_ID)
//      )
//      .selectExpr(T.getTBColumns: _*)
//    df.persist()
//    df.show(false)
//
//    df
//  }

  /**
   * 补充区域字段
   * @param data
   * @param day
   * @return
   */
  override def join(data: DataFrame, day: Day): DataFrame = {
    D.loadPeriodValidDevice(_DAY)
      .selectExpr(D.F_DEVICE_ID, D.F_REGION)
      .withColumnRenamed(D.F_DEVICE_ID, T.F_DEVICE_ID)
      .withColumnRenamed(D.F_REGION, T.F_REGION)
      // 确保新增设备，家庭，区域这些业务，能够跳过区域补充环节
      .join(data, Seq(T.F_DEVICE_ID), "right")
      // 转换消息类型
      .withColumn(T.F_SERVICE, col(T.F_SERVICE).cast(IntegerType))
  }


}
